package com.wang.thread;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;

public class EasyThreadPool {
    private int coreThreadSize;
    private int maxThreadSize;
    private Consumer<Runnable> reject;
    // 当前线程数
    private AtomicInteger threadSize = new AtomicInteger(0);
    // 空闲线程数
    private AtomicInteger waitingSize = new AtomicInteger(0);

    private BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(128);

    public EasyThreadPool(int coreThreadSize, int maxThreadSize, Consumer<Runnable> reject) {
        this.coreThreadSize = coreThreadSize;
        this.maxThreadSize = maxThreadSize;
        this.reject = reject;
    }

    public int getThreadSize() {
        return threadSize.intValue();
    }

    private void createWorker() {
        if(this.threadSize.incrementAndGet() > maxThreadSize) {
           this.threadSize.decrementAndGet();
           return ;
        }

        ThreadNode node = new ThreadNode();
        node.setTaskChannel(() -> {
            Runnable task = null;
            while ((task = this.queue.poll(5, TimeUnit.SECONDS)) == null) {
                synchronized (EasyThreadPool.this) {
                    // 双重判断，防止出现线程未完全进入等待状态时，有任务进来
                    if((task = this.queue.poll()) != null || waitingSize.intValue() > coreThreadSize) {
                        break;
                    }
                    waitingSize.incrementAndGet();
                    EasyThreadPool.this.wait();
                    waitingSize.decrementAndGet();
                }
            }

            if(task == null) {
                this.threadSize.decrementAndGet();
            }
            return task;
        });
        node.start();
    }

    public void addTask(Runnable task) {
        boolean isAdd = this.queue.add(task);
        // 过载执行拒绝策略
        if (!isAdd) {
            this.reject.accept(task);
            return;
        }

        if (this.threadSize.intValue() < maxThreadSize && this.queue.size() > 2) {
            this.createWorker();
        }

        // 唤醒闲置线程
        synchronized (this) {
            if (this.waitingSize.intValue() > 0) {
                this.notify();
            }
        }
    }
}
